package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/5/1 9:06
 */
public class DWD_06_DwdTradeOrderDetailApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_06_DwdTradeOrderDetailApp().init("DwdTradeOrderDetailApp", 3005, 1, "DwdTradeOrderDetailApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 读取订单预处理表
        tEnv.executeSql("" +
                            "create table dwd_trade_order_pre_process( " +
                            "id string," +
                            "order_id string," +
                            "user_id string," +
                            "order_status string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "date_id string," +
                            "create_time string," +
                            "operate_date_id string," +
                            "operate_time string," +
                            "source_id string," +
                            "source_type string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_total_amount string," +
                            "`type` string," +
                            "`old` map<string,string>," +
                            "od_ts string," +
                            "oi_ts string," +
                            "row_op_ts timestamp_ltz(3)" +
                            ")" + SQLUtil.getKafkaDDL("DwdTradeOrderDetailApp", "dwd_trade_order_pre_process"));
        
        //TODO 3.过滤出下单数据  type是新增即可
        Table filterTable = tEnv.sqlQuery("" +
                                              "select " +
                                              "id," +
                                              "order_id," +
                                              "user_id," +
                                              "sku_id," +
                                              "sku_name," +
                                              "province_id," +
                                              "activity_id," +
                                              "activity_rule_id," +
                                              "coupon_id," +
                                              "date_id," +
                                              "create_time," +
                                              "source_id," +
                                              "source_type source_type_code," +
                                              "source_type_name," +
                                              "sku_num," +
                                              "split_original_amount," +
                                              "split_activity_amount," +
                                              "split_coupon_amount," +
                                              "split_total_amount," +
                                              "od_ts ts," +
                                              "row_op_ts " +
                                              "from dwd_trade_order_pre_process " +
                                              "where `type`='insert'");
        tEnv.createTemporaryView("filter_table", filterTable);
        
        //TODO 4.创建Kafka下单数据表
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                            "id string," +
                            "order_id string," +
                            "user_id string," +
                            "sku_id string," +
                            "sku_name string," +
                            "province_id string," +
                            "activity_id string," +
                            "activity_rule_id string," +
                            "coupon_id string," +
                            "date_id string," +
                            "create_time string," +
                            "source_id string," +
                            "source_type_code string," +
                            "source_type_name string," +
                            "sku_num string," +
                            "split_original_amount string," +
                            "split_activity_amount string," +
                            "split_coupon_amount string," +
                            "split_total_amount string," +
                            "ts string," +
                            "row_op_ts timestamp_ltz(3)" +
                            ")" + SQLUtil.getKafkaDDL("", "dwd_trade_order_detail"));
        
        //TODO 5.将数据写出到Kafka
        tEnv.executeSql("insert into dwd_trade_order_detail select * from filter_table");
        
    }
}
